Serverless CacheサービスのMomentoがPub/Sub機能を実装します
Introduction
サーバレスキャッシュサービスMomentoの情報です。
最近機能追加がいろいろとされてますが、
（Preview版ながら）Pub/Sub機能が使えるようになりました。
今回はCLIおよびRust用SDKを使って
Pub/Sub機能を試してみます。
※2023年2月現在Pub/Sub機能はPreviewです
Environment
今回試した環境は以下のとおりです。
- MacBook Pro (13-inch, M1, 2020)
- OS : MacOS 12.4
- Rust : 1.66.1
- Momento CLI : v0.22.1
DevIOのMomento関連記事はここにあるので、
これらもご確認ください。
Setup
例によってMomentoのセットアップです。
このあたりを参考に、
MomentoのCLIインストールと認証トークンを取得しましょう。
認証トークンを取得したらコンソールでCLI用・プログラム用に
トークンをセットしておきます。
# CLI用 % momento configure Please paste your Momento auth token. (If you do not have an auth token, use `momento account` to generate one.) Windows users: if CTRL-V does not work, try right-click or SHIFT-INSERT to paste. Token [****]:<認証トークン> ・・・ #プログラム用 % export MOMENTO_AUTH_TOKEN = <取得した認証トークン>
Pub/Sub using CLI
ではMomento CLIでPub/Subの動作確認をしてみましょう。
% momento --version momento 0.29.0
Pub/Sub用のキャッシュを新たに作成します。
% momento cache create --name my_pubsub_cache
momento topic subscribeコマンドでSubscriberを起動します。
さきほど作成したキャッシュを指定し、任意の名前のトピック名を
指定してSubscribeます。
% momento topic subscribe --cache my_pubsub_cache my_topic
別のコンソールを起動し、↑で指定したキャッシュ・トピックに対してpublishします。
% momento topic publish --cache my_pubsub_cache my_topic 'hello' --verbose ・・・
Subscriberをみると、ちゃんとメッセージを受信できてます。
% momento topic subscribe --cache my_pubsub_cache my_topic hello #受信したメッセージ
CLIでtopicの動作確認ができたので、次はRustSDKで
PublisherとSubsriberを実装してみます。
Implement Pub/Sub using Rust
CLIで動作確認ができたので、
Rust SDKでも実装してみましょう。
Cargoで新規プロジェクトを作成します。
% cargo new momento_pubsub % cd momento_pubsub
Cargo.tomlを下記のように修正します。
MomentoライブラリやTokioなど、必要ライブラリの設定と、
binセクションでPublisherとSubscriber、
それぞれのプログラムを起動できるようにします。
# Cargo.toml
・・・
[dependencies]
momento = "0.22.1"
tokio = { version = "1.21.2", features = ["full"] }
anyhow = "1.0.66"
[[bin]]
name = "sub"
src = "src/bin/sub.rs"
[[bin]]
name = "pub"
src = "src/bin/pub.rs"
まずSubscriber（src/bin/sub.rs）の実装をしましょう。
キャッシュはさきほど作成したものを使うか、新しくcreateします。
// src/bin/sub.rs
use anyhow::self;
use std::env;
use momento::preview::topics::*;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let auth_token = env::var("MOMENTO_AUTH_TOKEN").unwrap();
let cache_name = "<Your demo Cache name for Topic>";
let topic_name = "my_topic";
let client = TopicClient::connect(auth_token,
None, Some("pubsub-demo")).expect("could not connect");
let mut subscription = client
.subscribe(cache_name.to_string(), topic_name.to_string(), None)
.await.expect("subscribe rpc failed");
while let Some(item) = subscription.item().await? {
match item {
SubscriptionItem::Value(value) => match value.kind {
ValueKind::Text(text) => println!("{text}"),
ValueKind::Binary(binary) => {
println!("{{\"kind\": \"binary\", \"length\": {}}}", binary.len())
}
},
SubscriptionItem::Discontinuity(discontinuity) => {
println!("{discontinuity:?}")
}
}
}
Ok(())
}
TopicClient::connectでクライアントを作成し、
キャッシュ名とトピック名を指定してsubscribe関数を実行します。
これでデータ待受の準備ができました、
なお、現状ではsubscribe実行後に60秒データがこないと
タイムアウトしてしまいます。
このあたりは今後改善される予定になってます。
次はPublisherの実装(src/bin/pub.rs)です。
TopicClient::connectでクライアントを作成し、
publish関数（もしくはpublish_mut）でデータのpublishを行います。
// src/bin/pub.rs
use anyhow::self;
use std::env;
use momento::preview::topics::*;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let auth_token = env::var("MOMENTO_AUTH_TOKEN").unwrap();
let cache_name = "demotopics";
let topic_name = "my_topic";
let client = TopicClient::connect(auth_token, None, Some("pubsub-demo"))?;
client.publish(cache_name.to_string(), topic_name.to_string(),
"Hello from Momento Publisher".to_string()).await?;
println!("publish done.");
Ok(())
}
では動作確認してみましょう。
まずはSubscriberを実行します。
% cargo run --bin sub
別ウィンドウでPublisherを実行します。
% cargo run --bin pub publish done
Subscriberのコンソールを確認すると、
メッセージを受信できているのがわかります。
% cargo run --bin sub hello from publisher
Summary
今回はMomentoの新機能であるPub/Sub機能を試してみました。
まだプレビュー版ですが、基本的な動作確認はできます。
ユーザー側で管理の必要がなく、
スケールも自動で大規模なシステムに対応も可能ですし、
完全従量課金（$0.15/GB)になっていますので、
Cache機能 ＆ Pub/Sub機能としてご検討ください。
なお、Momentoについてのお問い合わせはこちらです。
お気軽にお問い合わせください。
Momentoセミナーのお知らせ
2023年2月21日(火) 16:00からMomentoのセミナーを開催します。
興味があるかたはぜひご参加ください。
